package com.yunc.common.message;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yunc.common.redis.RedisService;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.PostConstruct;

/**
 * 消息监听器
 * @author taocongcong
 * @create 2018-10-30 11:48
 */
public abstract class AbstractMessageHandler<T> implements Runnable {

    private static final Logger logger = LogManager.getLogger(AbstractMessageHandler.class);

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;
    @Autowired
    private RedisService redisService;

    /**
     * 监听的消息类型
     */
    private String messageType;

    /**
     * 消息的class对象
     */
    private Class<T> clazz;

    private boolean monitor;

    /**
     * 默认为3
     */
    private int retryTimes = 3;

    public AbstractMessageHandler() {
    }

    public AbstractMessageHandler(String messageType, Class<T> clazz) {
        this.messageType = messageType;
        this.clazz = clazz;
    }

    @Override
    public void run() {
        while (true) {
            listen();
        }
    }

    /**
     * 初始化启动本监听器
     */
    @PostConstruct
    public void startListen() {
        // 启动监听
        new Thread(this).start();
    }

    public void listen() {
        // 获取队列中的值
        final Object obj = redisService.blpop(messageType, Integer.MAX_VALUE, Message.class);
        // 如果获取失败,以为着redis连接有问题
        if (obj == null) {
            monitor = false;
            logger.warn("消息分发器获取redis连接失败");
            // 暂停5秒钟,防止线程空跑,打印无数日志
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                logger.warn("消息分发器线程暂停失败");
            }
            return;
        }
        if (!monitor) {
            logger.warn("消息分发开始");
            monitor = true;
        }
        try {
            taskExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    Message message = (Message) obj;
                    // 如果获取成功，则交给子类做业务处理
                    // 转为bean
                    JSONObject jsonObj = (JSONObject) JSONObject.toJSON(message);
                    // JSON content = (JSON)jsonObj.get("content");
                    // JSONObject.parse(content, clazz);
                    T t = (T)JSONObject.toJavaObject((JSON) jsonObj.get("content"), clazz);
                    try {

                        handle(t);
                    } catch (Exception ex) {
                        logger.error(ex);
                        // 处理失败，判断是否需要重试
                        if (message.getFailTimes().intValue() >= retryTimes) {
                            handleFailed(t);
                        } else {
                            message.setFailTimes(message.getFailTimes().intValue() + 1);
                            // 再次put回消息总线，等待下次重试
                            redisService.rpush(messageType, message, 0);
                            if (logger.isDebugEnabled()) {
                                StringBuilder sb = new StringBuilder();
                                sb.append("msg:[").append(message).append("], 执行失败，准备重试。");
                                logger.debug(sb.toString());
                            }
                        }
                    }
                }
            });
        } catch (TaskRejectedException ex) {
            logger.warn("线程池已满，准备回写任务，暂停本线程");
            // 如果发生任务拒绝加入线程池，则回写任务到redis，等待重试
            redisService.rpush(messageType, obj, 0);
            // 暂停指定时间
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                logger.warn("生产者暂停异常", ex);
            }
        } catch (Exception ex) {
            logger.error("消息总线发生异常", ex);
        }
    }

    /**
     * 获取到消息后做业务处理
     */
    public abstract void handle(T obj);

    /**
     * 消息多次重试处理失败
     */
    public abstract void handleFailed(T obj);
}
